package flink_source;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.net.Socket;


/**
 * @author Spring_Hu
 * @date 2021/10/10 14:23
 */
public class Flink_userDefinedSource {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.addSource(new SourceFunction<Object>() {
            @Override
            public void run(SourceContext<Object> ctx) throws Exception {

                Socket socket = new Socket("hadoop102", 9999);
                BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream(), "utf8"));

                while(true){
                    String s = reader.readLine();
                    ctx.collect(s);
                }
            }

            @Override
            public void cancel() {

            }
        }).print();

        env.execute();

    }
}
